package com.demo.rocketmq.controller;

import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageClientExt;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;


/**
 * @author wangfengchen
 * 特别注意：重试的时候msgId是变化的，需要用key做唯一标识
 */
@Slf4j
public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("mpc_plain_short_message_group_local");
        consumer.setNamesrvAddr("rocketmq-12.test.newhuoidc.com:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("mpc_plain_short_message_topic_local", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                ConsumeConcurrentlyContext context) {

                long requestId = IdUtil.getSnowflakeNextId();
                log.info("requestId = {}，消费者收到消息，list = {}", requestId, list);
                for (MessageExt messageExt : list) {
//                    String msgId;
//                    // 此处注意当类型为MessageClientExt时，getMsgId()不是真的msgId，而是UniqueKey，所以需要调用getOffsetMsgId()方法
//                    if (messageExt instanceof MessageClientExt) {
//                        MessageClientExt ext = (MessageClientExt) messageExt;
//                        msgId = ext.getOffsetMsgId();
//                    } else {
//                        msgId = messageExt.getMsgId();
//                    }
                    String msg = new String(messageExt.getBody());
                    String topic = messageExt.getTopic();
                    String tags = messageExt.getTags();
                    String key = messageExt.getKeys();
                    int reconsume = messageExt.getReconsumeTimes();
                    // 如果达到重试次数不成功，则记录，人工补偿
                    if (reconsume >= 16) {
                        log.error("requestId = {}，消费者重试次数达到" + reconsume + "次，不再重试，返回成功，人工补偿", requestId);
                        // 达到最大次数后人工补偿，从redis中拿出失败消息
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    log.info("requestId = {}，消息内容，key = {}, msg = {}，topic = {}，tags = {}", requestId, key, msg, topic, tags);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started...%n");
    }
}
